//package com.example.service;
//
//import com.example.Utils.Env;
//import com.example.entity.StockTrade;
//import jakarta.annotation.PostConstruct;
//import jakarta.annotation.Resource;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.functions.AggregateFunction;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
//import org.springframework.stereotype.Component;
//import redis.clients.jedis.Jedis;
//@Component
//@Slf4j
//public class TopTenPlatform {
//    @Resource
//    private Env env_;
//
//    @PostConstruct
//    public void init() {
//        new Thread(() -> {
//            try {
//                startFlinkJob();
//            } catch (Exception e) {
//                log.error("Error starting Flink job", e);
//            }
//        }).start();
//    }
//
//    public  void startFlinkJob() throws Exception {
//        // 创建 Kafka source
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        DataStream<StockTrade> tradeStream = env_.getEnv(env);
//
//        log.info("数据源搞好了");
//
//        // 1 小时窗口
//        tradeStream
//                .map(new MapFunction<StockTrade, Tuple2<String, Long>>() {
//                    @Override
//                    public Tuple2<String, Long> map(StockTrade value) throws Exception {
//                        return new Tuple2<>(value.getTradePlatform(), value.getTradeVolume());
//                    }
//                })
//                .keyBy(value -> value.f0)
//                .window(TumblingEventTimeWindows.of(Time.hours(1)))
//                .aggregate(new TradeAggregator())
//                .addSink(new AccumulatingRedisSink(redisConfig));
//
//        System.out.println("Top 10 stock trade volume analysis started");
//        env.execute("Stock Trade Analysis");
//    }
//
//    public FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
//            .setHost("localhost")
//            .setPort(6379)
//            .build();
//
//    // 自定义 Sink 函数
//// 自定义 Sink 函数
//    public static class AccumulatingRedisSink extends RichSinkFunction<Tuple2<String, Long>> {
//        private transient Jedis jedis;
//        private final FlinkJedisPoolConfig config;
//
//        public AccumulatingRedisSink(FlinkJedisPoolConfig config) {
//            this.config = config;
//        }
//
//        @Override
//        public void open(Configuration parameters) throws Exception {
//            super.open(parameters);
//            jedis = new Jedis(config.getHost(), config.getPort());
//        }
//
//        @Override
//        public void close() throws Exception {
//            if (jedis != null) {
//                jedis.close();
//            }
//            super.close();
//        }
//
//        @Override
//        public void invoke(Tuple2<String, Long> value, Context context) throws Exception {
//
//            String hashKey = "stock_trade_platform"; // 存储股票交易量的键
//            String field = value.f0; // 使用股票平台作为字段
//
//            // 检查键的类型
//            String type = jedis.type(hashKey);
//            if (!"zset".equals(type)) {
//                // 如果键不存在或类型不正确，删除旧的键
//                jedis.del(hashKey);
//                // 创建新的有序集合
//                jedis.zadd(hashKey, 0, field); // 初始化一个空的有序集合
//            }
//
//            // 从 Redis 中获取现有值
//            Double existingVolume = jedis.zscore(hashKey, field);
//            if (existingVolume == null) {
//                existingVolume = (double) 0L;
//            }
//
//            // 更新交易量
//            double updatedVolume = existingVolume + value.f1;
//
//            // 将更新后的值写回 Redis
//            jedis.zadd(hashKey, updatedVolume, field);
//            final long delayInterval = 100;
//            Thread.sleep(delayInterval);
//        }
//
//    }
//
//
//    // 定义聚合函数
//    public static class TradeAggregator implements AggregateFunction<Tuple2<String, Long>, Tuple2<String, Long>, Tuple2<String, Long>> {
//
//
//        @Override
//        public Tuple2<String, Long> createAccumulator() {
//            return new Tuple2<>("", 0L);
//        }
//
//        @Override
//        public Tuple2<String, Long> add(Tuple2<String, Long> stringDoubleTuple2, Tuple2<String, Long> stringDoubleTuple22) {
//            String stockCode = stringDoubleTuple2.f0;
//            Long totalAmount = stringDoubleTuple2.f1 + stringDoubleTuple22.f1;
//            return new Tuple2<>(stockCode, totalAmount);
//        }
//
//
//
//
//        @Override
//        public Tuple2<String, Long> getResult(Tuple2<String, Long> accumulator) {
//            return accumulator;
//        }
//
//        @Override
//        public Tuple2<String, Long> merge(Tuple2<String, Long> a, Tuple2<String, Long> b) {
//            String stockCode = a.f0; // 假设两个 accumulator 的股票代码相同
//            Long totalAmount = a.f1 + b.f1;
//            return new Tuple2<>(stockCode, totalAmount);
//        }
//
//
//    }
//}
